package b.utils.entity

import akka.Done
import com.datastax.driver.core.{BoundStatement, PreparedStatement}
import com.lightbend.lagom.scaladsl.persistence.ReadSideProcessor.ReadSideHandler
import com.lightbend.lagom.scaladsl.persistence.cassandra.{CassandraReadSide, CassandraSession}
import com.lightbend.lagom.scaladsl.persistence.{AggregateEventTag, EventStreamElement, ReadSideProcessor}

import scala.collection.immutable.Seq
import scala.concurrent.{ExecutionContext, Future}

class C事务ReadSide(
                 readSide: CassandraReadSide,
                 session: CassandraSession)(implicit ec: ExecutionContext) extends ReadSideProcessor[T事务Event]{

  @volatile private var v事务Create: PreparedStatement = _ // initialized in prepare
  @volatile private var v事务Update: PreparedStatement = _ // initialized in prepare
  @volatile private var v事务参与方Create: PreparedStatement = _ // initialized in prepare
  @volatile private var v事务参与方Update: PreparedStatement = _ // initialized in prepare

  def f建表(): Future[Done] = {
    val transaction = s"""
                         |CREATE TABLE IF NOT EXISTS tcc (
                         |  sequence bigint,
                         |  name varchar,
                         |  participants text,
                         |  status smallint,
                         |  PRIMARY KEY (sequence)
                         |)""".stripMargin
    val participant = s"""
                         |CREATE TABLE IF NOT EXISTS tcc_participant (
                         |  sequence bigint,
                         |  participant varchar,
                         |  participant_entity_id varchar,
                         |  status smallint,
                         |  PRIMARY KEY (sequence, participant)
                         |)""".stripMargin
    Future.sequence(Seq[Future[Done]](
      session.executeCreateTable(transaction),
      session.executeCreateTable(participant)
    )).map(_ => Done)
  }

  def fSQL语句: Future[Done] = {
    Future.sequence(Seq[Future[Done]](
      session.prepare("INSERT INTO tcc (sequence, name, participants, status) VALUES (?, ?, ?, 0)")
        .map { ps => v事务Create = ps; Done},
      session.prepare("UPDATE tcc SET status = ? WHERE sequence = ?")
        .map { ps => v事务Update = ps; Done},
      session.prepare("INSERT INTO tcc_participant (sequence, participant, participant_entity_id, status) VALUES (?, ?, ?, 0)")
        .map { ps => v事务参与方Create = ps; Done},
      session.prepare("UPDATE tcc_participant SET status = ? WHERE sequence = ? AND participant = ?")
        .map { ps => v事务参与方Update = ps; Done}
    )).map(_ => Done)
  }

  private def f新事务(event: CC事务Started): Future[Seq[BoundStatement]] = {
    println(s"f新事务: $event")
    Future.successful({
      val stmt = v事务Create.bind()
      stmt.setLong("sequence", event.v事务ID)
      stmt.setString("name", event.v事务名称)
      stmt.setString("participants", event.v参与方EntitesID.keys.mkString(","))
      val stmts = event.v参与方EntitesID map { p =>
        val stmt1 = v事务参与方Create.bind()
        stmt1.setLong("sequence", event.v事务ID)
        stmt1.setString("participant", p._1)
        stmt1.setString("participant_entity_id", p._2)
        stmt1
      }
      Seq(stmt) ++ stmts
    })
  }

  private def f事务提交(event: CC事务Committed): Future[Seq[BoundStatement]] = {
    println(s"f事务提交: $event")
    Future.successful({
      val stmt = v事务Update.bind()
      stmt.setShort("status", 1.toShort)
      stmt.setLong("sequence", event.v事务ID)
      Seq(stmt)
    })
  }

  private def f事务取消(event: CC事务RolledBack): Future[Seq[BoundStatement]] = {
    println(s"f事务取消: $event")
    Future.successful({
      val stmt = v事务Update.bind()
      stmt.setShort("status", (-1).toShort)
      stmt.setLong("sequence", event.v事务ID)
      Seq(stmt)
    })
  }

  private def f参与方尝试(event: CC事务参与方Tried): Future[Seq[BoundStatement]] = {
    println(s"f参与方尝试: $event")
    Future.successful({
      val stmt = v事务参与方Update.bind()
      stmt.setShort("status", 1.toShort)
      stmt.setLong("sequence", event.v事务ID)
      stmt.setString("participant", event.v参与方)
      Seq(stmt)
    })
  }

  private def f参与方确认(event: CC事务参与方Confirmed): Future[Seq[BoundStatement]] = {
    println(s"f参与方确认: $event")
    Future.successful({
      val stmt = v事务参与方Update.bind()
      stmt.setShort("status", 2.toShort)
      stmt.setLong("sequence", event.v事务ID)
      stmt.setString("participant", event.v参与方)
      Seq(stmt)
    })
  }

  private def f参与方取消(event: CC事务参与方Canceled): Future[Seq[BoundStatement]] = {
    println(s"f参与方取消: $event")
    Future.successful({
      val stmt = v事务参与方Update.bind()
      stmt.setShort("status", (-1).toShort)
      stmt.setLong("sequence", event.v事务ID)
      stmt.setString("participant", event.v参与方)
      Seq(stmt)
    })
  }

  override def aggregateTags: Set[AggregateEventTag[T事务Event]] = {
    Set(O事务Event.TAG)
  }

  override def buildHandler(): ReadSideHandler[T事务Event] = {
    val b = readSide.builder[T事务Event]("TCC")
    b.setGlobalPrepare(f建表)
    b.setPrepare(_ => fSQL语句)
    b.setEventHandler[CC事务Started]((c: EventStreamElement[CC事务Started]) => f新事务(c.event))
    b.setEventHandler[CC事务Committed]((c: EventStreamElement[CC事务Committed]) => f事务提交(c.event))
    b.setEventHandler[CC事务RolledBack]((c: EventStreamElement[CC事务RolledBack]) => f事务取消(c.event))
    b.setEventHandler[CC事务参与方Tried]((c: EventStreamElement[CC事务参与方Tried]) => f参与方尝试(c.event))
    b.setEventHandler[CC事务参与方Confirmed]((c: EventStreamElement[CC事务参与方Confirmed]) => f参与方确认(c.event))
    b.setEventHandler[CC事务参与方Canceled]((c: EventStreamElement[CC事务参与方Canceled]) => f参与方取消(c.event))
    b.build()
  }
}

